package org.apache.ocean;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.apache.commons.lang.StringUtils;
import org.apache.lucene.analysis.Analyzer;
import org.apache.lucene.document.Document;
import org.apache.lucene.index.CorruptIndexException;
import org.apache.lucene.index.IndexCommitPoint;
import org.apache.lucene.index.IndexDeletionPolicy;
import org.apache.lucene.index.IndexReader;
import org.apache.lucene.index.IndexWriter;
import org.apache.lucene.index.MergePolicy;
import org.apache.lucene.index.SegmentInfos;
import org.apache.lucene.index.SerialMergeScheduler;
import org.apache.lucene.index.StaleReaderException;
import org.apache.lucene.index.Term;
import org.apache.lucene.index.TermEnum;
import org.apache.lucene.store.Directory;
import org.apache.lucene.store.IndexInput;
import org.apache.lucene.store.IndexOutput;
import org.apache.lucene.store.LockObtainFailedException;
import org.apache.ocean.snapshot.Snapshot;
import org.apache.ocean.snapshot.Snapshot.SnapshotFile;

/**
 * TODO: unit test ClusterIndexModifier
 * 
 * @author jasonr
 */
// TODO: use existing SegmentReader in indexWriter.applyDeletes
// TODO: enable batch updates by creating field in transactionrecord and
// sequence for a batch id
// TODO: adjust term index interval
public class ClusterIndexModifier {
	public static Logger log = Logger.getLogger(ClusterIndexModifier.class.getName());
	public static final long TRY_LOCK_DURATION = 1000 * 3;
	private Directory directory;
	private Analyzer analyzer;
	private IndexWriter indexWriter;
	private IndexReader indexReader;
	private AtomicLong docsAddedSinceReload = new AtomicLong(0);
	private AtomicLong docsDeletedSinceReload = new AtomicLong(0);
	private AtomicLong docsAdded = new AtomicLong(0);
	private AtomicLong docsDeleted = new AtomicLong(0);
	private AtomicLong reloadCount = new AtomicLong(0);
	private Date dateOpened;
	private ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
	private boolean isClosed = false;
	private boolean addedIndices = false;
	ClusterIndexDeletionPolicy clusterIndexDeletionPolicy;

	public ClusterIndexModifier(CDouble ramBufferSize, MergePolicy mergePolicy, Directory directory, Analyzer analyzer) throws CorruptIndexException, IOException {
		this.directory = directory;
		this.analyzer = analyzer;
		boolean createIndex = !IndexReader.indexExists(directory);
		clusterIndexDeletionPolicy = new ClusterIndexDeletionPolicy(1000*60*2);
		indexWriter = new IndexWriter(directory, true, analyzer, createIndex, clusterIndexDeletionPolicy);
		if (ClusterUtil.debugIndexWriter())
			indexWriter.setInfoStream(System.out);
		if (mergePolicy != null)
			indexWriter.setMergePolicy(mergePolicy);
		if (ramBufferSize != null)
			indexWriter.setRAMBufferSizeMB(ramBufferSize.value);
		indexWriter.segmentReaderLoader = new ClusterSegmentReaderLoader();
		indexWriter.setWriteLockTimeout(SearchIndexShard.WRITE_LOCK_TIMEOUT);
		indexWriter.setMergeScheduler(new SerialMergeScheduler());
		indexWriter.setMaxBufferedDeleteTerms(20000);
		// indexWriter.setTermIndexInterval(interval)
		indexReader = IndexReader.open(directory);
		dateOpened = new Date();
	}

	public Snapshot getLatestSnapshot() throws Exception {
		return clusterIndexDeletionPolicy.getLatestSnapshot();
	}
  
	public void renewSnapshotLease(String snapshotName) {
		if (snapshotName == null) return;
		InternalSnapshot internalSnapshot = clusterIndexDeletionPolicy.map.get(snapshotName);
		if (internalSnapshot != null) {
			internalSnapshot.renewLease();
		}
	}
	
	public class InternalSnapshot {
		IndexCommitPoint indexCommitPoint;
		long timestamp;

		public InternalSnapshot(IndexCommitPoint indexCommitPoint) {
			this.indexCommitPoint = indexCommitPoint;
			timestamp = System.currentTimeMillis();
		}
		
		public void renewLease() {
			timestamp = System.currentTimeMillis();
		}
		
		public Snapshot getSnapshot() throws Exception {
			Snapshot snapshot = new Snapshot();
			snapshot.name = indexCommitPoint.getSegmentsFileName();
			snapshot.timestamp = new Date(timestamp);
			Collection<String> fileNames = indexCommitPoint.getFileNames();
			for (String fileName : fileNames) {
				snapshot.add(toSnapshotFile(fileName));
			}
			return snapshot;
		}
		
		private SnapshotFile toSnapshotFile(String fileName) throws Exception {
			SnapshotFile snapshotFile = new SnapshotFile();
			snapshotFile.lastModified = new CLong(directory.fileModified(fileName));
			snapshotFile.size = new CLong(directory.fileLength(fileName));
			snapshotFile.name = fileName;
			return snapshotFile;
		}
	}

	public class ClusterIndexDeletionPolicy implements IndexDeletionPolicy {
		TreeMap<CLong, InternalSnapshot> map = new TreeMap<CLong, InternalSnapshot>();
		ReentrantLock lock = new ReentrantLock();
		long oldestSnapshotDuration;
    
		public ClusterIndexDeletionPolicy(long oldestSnapshotDuration) {
			this.oldestSnapshotDuration = oldestSnapshotDuration;
		}
		
		public Snapshot getLatestSnapshot() throws Exception {
			lock.lock();
			try {
				if (map.size() == 0) return null;
				InternalSnapshot internalSnapshot = map.get(map.lastKey());
				return internalSnapshot.getSnapshot();
			} finally {
				lock.unlock();
			}
		}
    // TODO: add lastest snapshot to getshardinfos
		public void onInit(List indexCommitPoints) throws IOException {
		  onCommit(indexCommitPoints);
		}

		public void onCommit(List indexCommitPoints) throws IOException {
			lock.lock();
			try {
				if (indexCommitPoints.size() == 0) return; 
				IndexCommitPoint indexCommitPoint = (IndexCommitPoint) indexCommitPoints.get(indexCommitPoints.size()-1);
				String segmentsFileName = indexCommitPoint.getSegmentsFileName();
				log.info("segmentsFileName: "+segmentsFileName+" map.size: "+map.size());
				if (StringUtils.equals("segments.gen", segmentsFileName)) {
					log.log(Level.SEVERE, "cannot handle "+segmentsFileName);
					throw new RuntimeException("cannot handle "+segmentsFileName);
				}
				long generation = SegmentInfos.generationFromSegmentsFileName(segmentsFileName);
				map.put(new CLong(generation), new InternalSnapshot(indexCommitPoint));
				cleanNoLock();
			} finally {
				lock.unlock();
			}
		}

		private void cleanNoLock() {
			Iterator<Map.Entry<CLong, InternalSnapshot>> iterator = map.entrySet().iterator();
			while (iterator.hasNext()) {
				Map.Entry<CLong, InternalSnapshot> entry = iterator.next();
				InternalSnapshot internalSnapshot = entry.getValue();
				long age = System.currentTimeMillis() - internalSnapshot.timestamp;
				if (age > oldestSnapshotDuration) {
					internalSnapshot.indexCommitPoint.delete();
					log.info("delete "+internalSnapshot.indexCommitPoint.getSegmentsFileName());
					map.remove(entry.getKey());
				}
			}
		}
	}
  
	public IndexOutput getIndexFileOutput(String fileName) throws Exception {
		return directory.createOutput(fileName);
	}
	
	public IndexInput getIndexFileInput(String snapshotName, String fileName) throws IOException {
		if (!directory.fileExists(fileName))
			return null;
		renewSnapshotLease(snapshotName);
		return directory.openInput(fileName);
	}

	public String getID() {
		return directory.getLockID();
	}

	public class ClusterSegmentReaderLoader implements IndexWriter.SegmentReaderLoader {
		public IndexReader getSegmentReader(String segmentName) {
			try {
				IndexReader segmentIndexReader = ClusterUtil.getIndexReaderBySegmentName(segmentName, indexReader);
				// log.info("segmentIndexReader: "+segmentIndexReader);
				return segmentIndexReader;
				// return null;
			} catch (Throwable throwable) {
				log.log(Level.SEVERE, "", throwable);
				return null;
			}
		}
	}

	public CLong getSegmentsVersion() {
		return new CLong(indexReader.getVersion());
	}

	public CLong getReloadCount() {
		return new CLong(reloadCount.get());
	}

	public List<SegmentIndexSearcher> createSegmentIndexSearchers() throws Exception {
		errorIfClosed();
		return createMultiIndexSearcher().getSegmentIndexSearchers();
	}

	public Directory getDirectory() {
		return directory;
	}

	public IndexReader getIndexReader() {
		return indexReader;
	}

	public void readLock() {
		lock.readLock().lock();
	}

	public void readUnlock() {
		lock.readLock().unlock();
	}

	public void close() throws IOException {
		isClosed = true;
		lock.writeLock().lock();
		try {
			try {
				indexWriter.close();
			} finally {
				indexReader.close();
			}
		} finally {
			lock.writeLock().unlock();
		}
	}

	public Term getTerm(Term startingTerm) throws CorruptIndexException, IOException {
		readLock();
		try {
			List<Term> terms = getTerms(startingTerm, 1);
			if (terms.size() == 0)
				return null;
			else
				return terms.get(0);
		} finally {
			readUnlock();
		}
	}

	public List<Term> getTerms(Term startingTerm, int length) throws CorruptIndexException, IOException {
		List<Term> results = new ArrayList<Term>(length);
		TermEnum termEnum = indexReader.terms(startingTerm);
		String field = startingTerm.field();
		int count = 0;
		try {
			do {
				Term term = termEnum.term();
				if (term == null || term.field() != field)
					break;
				results.add(term);
				count++;
			} while (termEnum.next() && count < length);
		} finally {
			if (termEnum != null) {
				termEnum.close();
			}
		}
		return results;
	}

	public void optimize() throws CorruptIndexException, IOException {
		errorIfClosed();
		indexWriter.optimize();
	}

	public void addIndexesNoOptimize(Directory[] directories, IndexWriter.TransactionSemantics transactionSemantics) throws CorruptIndexException, IOException {
		errorIfClosed();
		indexWriter.addIndexesNoOptimize(directories, transactionSemantics);
		indexWriter.flush();
		addedIndices = true;
	}

	public MultiIndexSearcher createMultiIndexSearcher() throws IOException {
		errorIfClosed();
		return new MultiIndexSearcher(indexReader);
	}

	public Date getDateOpened() {
		return dateOpened;
	}

	protected void errorIfClosed() throws IOException {
		if (isClosed)
			throw new IOException("closed");
	}

	public void addDocuments(List<Document> documents, Analyzer analyzer, boolean flush) throws CorruptIndexException, IOException, InterruptedException,
			Exception {
		for (Document document : documents) {
			addDocument(document, analyzer, false);
		}
		if (flush)
			indexWriter.flush();
	}

	public void addDocument(Document document, Analyzer analyzer, boolean flush) throws CorruptIndexException, IOException, InterruptedException, Exception {
		errorIfClosed();
		indexWriter.addDocument(document, analyzer);
		docsAddedSinceReload.incrementAndGet();
		docsAdded.incrementAndGet();
		if (flush)
			indexWriter.flush();
	}

	public CLong count(Term term) throws IOException {
		return new CLong(indexReader.docFreq(term));
	}

	/**
	 * for consistency a reload should happen before this is called to insure
	 * something is actually deleted. if it does not exist then it will not be
	 * deleted
	 * 
	 * @param term
	 * @param ifDeletedReload
	 * @return
	 * @throws StaleReaderException
	 * @throws CorruptIndexException
	 * @throws LockObtainFailedException
	 * @throws IOException
	 * @throws Exception
	 */
	public CLong deleteDocuments(Term term, boolean reloadBeforeDelete, boolean flush, boolean reloadAfterFlush) throws StaleReaderException,
			CorruptIndexException, LockObtainFailedException, IOException, Exception {
		errorIfClosed();
		if (reloadBeforeDelete)
			reload();
		CLong count = count(term);
		if (count.value == 0)
			return new CLong(0);
		indexWriter.deleteDocuments(term);
		docsDeletedSinceReload.addAndGet(count.value);
		docsDeleted.addAndGet(count.value);
		if (flush)
			indexWriter.flush();
		if (reloadAfterFlush && flush) {
			reload();
		}
		return count;
	}

	public CDouble getSizeMB() throws IOException {
		long total = 0;
		for (String name : directory.list()) {
			total += directory.fileLength(name);
		}
		return new CDouble(total / (1024 * 1024));
	}

	public void reload() throws CorruptIndexException, IOException, Exception {
		reload((ReloadCall) null);
	}

	public void reload(ReloadCall reloadCall) throws CorruptIndexException, IOException, Exception {
		errorIfClosed();
		lock.writeLock().lock();
		try {
			if (indexReader == null) {
				throw new Exception("indexReader null");
			}
			if (indexWriter == null) {
				throw new Exception("indexWriter null");
			}
			indexWriter.flush();
			if (addedIndices || docsDeletedSinceReload.get() > 0 || docsAddedSinceReload.get() > 0) {
				// log.info("reload");
				IndexReader newIndexReader = indexReader.reopen();
				if (reloadCall != null)
					reloadCall.beforeNewReaderSet();
				this.indexReader = newIndexReader;
				reloadCount.incrementAndGet();
			}
			docsAddedSinceReload.set(0);
			docsDeletedSinceReload.set(0);
			addedIndices = false;
		} finally {
			lock.writeLock().unlock();
		}
	}
}
